package com.bw.func;

import com.bw.app.bean1.DwsPopularityMetricbean;
import com.bw.app.bean1.DwsUserBehaviorbean;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.java.tuple.Tuple4;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.tuple.Tuple4;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

/**
 * @projectName: Flinkgd
 * @fileName: PopularityProcessFunc
 * @packageName: com.bw.func
 * @date: 2025/9/23 22:40
 * @author: ymw
 */
public class PopularityProcessFunc extends KeyedProcessFunction<
        Tuple4<String, String, String, String>, // key 类型
        DwsUserBehaviorbean,                    // 输入
        DwsPopularityMetricbean> {
    // 计数器状态
    private ValueState<Long> counterState;
    // 定时器触发时间状态
    private ValueState<Long> triggerTimeState;

    @Override
    public void open(Configuration parameters) {
        counterState = getRuntimeContext().getState(
                new ValueStateDescriptor<>("counter", Long.class, 0L));
        triggerTimeState = getRuntimeContext().getState(
                new ValueStateDescriptor<>("trigger-time", Long.class));
    }

    @Override
    public void processElement(DwsUserBehaviorbean value,
                               Context ctx,
                               Collector<DwsPopularityMetricbean> out) throws Exception {

        // 1. 计数 +1
        long cnt = counterState.value() + 1;
        counterState.update(cnt);

        // 2. 第一次到达时注册 5 分钟定时器
        if (triggerTimeState.value() == null) {
            long trigger = ctx.timerService().currentProcessingTime() + 5 * 60 * 1000;
            ctx.timerService().registerProcessingTimeTimer(trigger);
            triggerTimeState.update(trigger);
        }
    }

    @Override
    public void onTimer(long timestamp,
                        OnTimerContext ctx,
                        Collector<DwsPopularityMetricbean> out) throws Exception {

        // ✅ 这里可以合法调 getCurrentKey()
        Tuple4<String, String, String, String> key = ctx.getCurrentKey();
        long cnt = counterState.value();
        long end = timestamp;
        long start = end - 5 * 60 * 1000;

        // 3. 向下游发送结果
        out.collect(new DwsPopularityMetricbean(
                key.f0, key.f1, key.f2, key.f3,
                start, end, cnt));

        // 4. 清理状态，下一个窗口重新计数
        counterState.clear();
        triggerTimeState.clear();
    }
}
